use crate::contract::coolink_sdk::{P2PRPCServiceCallback, P2PRPCServiceContract, SDKApiResult};
use crate::imp::tonicrpc::tonicserver::{
    _get_tonic_client, cancel_bytebuf_service_duplex, cancel_content_service_duplex,
    close_remote_tonic_tunnel, get_or_set_bytebuffer_callback, get_or_set_bytebuffer_sender,
    get_or_set_content_sender, get_or_set_remote_content_callback,
    get_or_set_service_command_sender, get_or_set_tunnel_client_command_sender, get_tunnel_creator,
    is_tonic_server_start, request_gather_candidates, request_send_bytes, request_send_content,
    select_bytebuf_service_duplex, select_content_service_duplex, set_tonic_server_start,
    TonicServer, TunnelClientCommand, TunnelServiceCommand,
};
use anyhow::anyhow;

use crate::imp::tonicrpc::kcptunnel::tunnel::URIInfo;
use crate::imp::tonicrpc::pairinlanmap;
use crate::imp::tonicrpc::tonicservercallback::CustomTonicServerCallback;
use crate::imp::tonicrpc::tonicserverimp::operate_download_path_map;
use crate::imp::tonicrpc::tonictunnel::{
    handle_selected_bytebuffer_duplex, handle_selected_contents_duplex, TonicTunnelCallback,
};
use crate::model::{
    sdk_api_result::{
        SDKResultError, DEFAULT_CONNECTION_TIMEOUT_SECS, DEFAULT_REQUEST_TIMEOUT_SECS,
    },
    sdk_req_result::ReqResult,
    storage_config::{
        OptionalP2PRPCLocalStorageConfig, P2PRPCLocalStorageConfig, APP_INFO,
        P2PRPC_LOCAL_CONFIG_FILE_NAME,
    },
};

use optional_struct::optional_struct;
use optional_struct::Applyable;

use serde::{Deserialize, Serialize};
use state::Storage;
use tokio::runtime::Builder;

use crate::utils::{ids::id_generate_xbit, logutil, preferences::Preferences, query, time_utils};
use std::fs;
use std::fs::{File, OpenOptions};
use std::io;
use std::io::prelude::*;
use std::path::PathBuf;
use std::sync::RwLock;
use std::time::Duration;

use futures_channel::mpsc::UnboundedReceiver;
use futures_io::SeekFrom;
use futures_util::{future::Either, pin_mut, SinkExt, StreamExt};
use url::Url;

use tonic_inf::api::{
    byte_buffer_msg::Bfmsg, send_content_msg::Msg, ByteBufferStop, SendContentStop,
};

lazy_static! {
    static ref P2PRTC_INSTANCE: Storage<RwLock<P2PRPCServiceContractImp>> = state::Storage::new();
}

// set sdk instance only once
fn set_sdk_instance(rtm_sdk: P2PRPCServiceContractImp) -> bool {
    // if try get success from state::Storage
    let ok = if let Some(_) = P2PRTC_INSTANCE.try_get() {
        // if sdk instance call many times in many thread and this api will return false.
        log::warn!("rtm sdk instance already inited!");
        false
    } else {
        // set new connect_state
        P2PRTC_INSTANCE.set(RwLock::new(rtm_sdk));
        true
    };
    ok
}

/// TurnAddrData from turn server
#[derive(Serialize, Deserialize)]
struct TurnAddrData {
    // turn group address
    #[serde(rename = "turn_group_address")]
    turn_group_address: String,
    // turn user name
    #[serde(rename = "turn_user_name")]
    turn_user_name: String,
    // turn user pwd
    #[serde(rename = "turn_user_pwd")]
    turn_user_pwd: String,
}

/// P2PRPCConfigOptions option struct need by p2prpc server initalization
#[optional_struct]
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct P2PRPCConfig {
    pub device_package: String,
    pub device_id: String,
    pub user_id: String,
    pub client_id: String,

    pub turn_uri: String,
    pub turn_usr: String,
    pub turn_pwd: String,
    pub turn_ttl: u32,
}

impl P2PRPCConfig {
    fn default() -> Self {
        Self {
            device_package: "".to_owned(),
            device_id: "".to_owned(),
            user_id: "".to_owned(),
            client_id: "".to_owned(),

            turn_uri: "".to_owned(),
            turn_usr: "".to_owned(),
            turn_pwd: "".to_owned(),
            turn_ttl: 300,
        }
    }

    fn from_json(json: &str) -> anyhow::Result<Self> {
        let mut info = P2PRPCConfig::default();
        if !json.eq("") {
            let option = serde_json::from_str::<OptionalP2PRPCConfig>(json)?;
            option.apply_to(&mut info);
        }
        Ok(info)
    }
}

pub struct P2PRPCServiceContractImp;

impl P2PRPCServiceContractImp {
    fn new() -> Self {
        P2PRPCServiceContractImp {}
    }

    async fn refresh_turn_addr(
        &self,
        client: &reqwest::Client,
        server_host: &String,
        subgroup_path: &String,
        app_key: &String,
        secret: &String,
        device_id: &String,
        device_pkg: &String,
        user_id: &String,
    ) -> Result<TurnAddrData, anyhow::Error> {
        if device_id == "" || server_host == "" || device_pkg == "" {
            let msg = "device_id or host or device_package is empty!";
            log::info!("{}", msg);
            anyhow::bail!(msg);
        }
        if app_key == "" || secret == "" {
            let msg = "app_key or secret empty!";
            log::info!("{}", msg);
            anyhow::bail!(msg);
        }
        log::info!(
            "turn_group_host: {} ,device_id:{},device_pkg:{}",
            server_host,
            device_id,
            device_pkg
        );

        let mut query: Vec<(String, String)> = vec![
            ("device_id".to_owned(), device_id.to_owned()),
            ("device_pkg".to_owned(), device_pkg.to_owned()),
            ("uid".to_owned(), user_id.to_owned()),
        ];
        let query_str = query::assemble_query_config((app_key, secret), &mut query)?;

        let has_prefix =
            if server_host.starts_with("http://") || server_host.starts_with("https://") {
                true
            } else {
                false
            };
        let turn_group_url = if has_prefix {
            format!("{}{}?{}", server_host, subgroup_path, query_str)
        } else {
            format!("https://{}{}?{}", server_host, subgroup_path, query_str)
        };

        log::info!("query_turn_url: {}", turn_group_url);
        // if url is not a illegal url ,then return.
        Url::parse(&turn_group_url)?;

        let res = client
            .get(&turn_group_url)
            .timeout(Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_SECS))
            .send()
            .await?;

        let body = res.text().await?;
        let ReqResult { code, data, msg } = serde_json::from_str(&body)?;
        if code == 0 {
            match data {
                Some(turn_data) => {
                    let turn_addr_data = serde_json::from_value(turn_data)?;
                    Ok(turn_addr_data)
                }
                _ => anyhow::bail!("query_turn_url fail,code code = {}", code),
            }
        } else {
            if msg.is_some() {
                anyhow::bail!(
                    "query_turn_url fail,msg.is_some() code = {},msg = {}",
                    code,
                    msg.unwrap()
                );
            } else {
                anyhow::bail!("query_turn_url fail,msg.is_none() code = {}", code);
            }
        }
    }

    fn sync_start_tonic_server(
        &self,
        sg_host: String,          // gateway query clientid url host address
        sg_subgroup_path: String, //  gateway query clientid url subgroup path
        sg_access_key: String,    // gateway registed http request auth access key
        sg_access_secret: String, // gateway registed http request auth access secret,corresponding for access key

        config_file_saved_path: String,    // config file saved path
        download_files_saved_path: String, // download files saved path
        enable_httpd_callback: bool,       // check if enable httpd callback
        p2prpc_callback: Box<dyn P2PRPCServiceCallback>, //p2prpc call back dyn point
        ice_turn_config_json: String,      // ice turn config json string, mapping to TurnAddrs
    ) -> SDKApiResult {
        if let Ok(rt) = Builder::new_multi_thread()
            .on_thread_start(|| {
                // call_attach_fn();
            })
            .on_thread_stop(|| {
                // call_detach_fn();
            })
            .enable_all()
            .build()
        {
            // block on tokio async loop. sync_connect must be called in a thread!
            let res = rt.block_on(self.async_start_tonic_server(
                sg_host,
                sg_subgroup_path,
                sg_access_key,
                sg_access_secret,
                config_file_saved_path,
                download_files_saved_path,
                enable_httpd_callback,
                p2prpc_callback,
                ice_turn_config_json,
            ));

            log::info!("sync_start_tonic_server run end res: {:?}", &res);
            if !set_tonic_server_start(false) {
                log::warn!("sync_start_tonic_server exit blockon,but set_connect_state return false and is_tonic_server_start = {:?}!",is_tonic_server_start());
            }
            // quit tonic server beacuse close or some err,
            // so ,our can retry to
            if let Err(e) = res {
                // -1 mean library error!;
                log::info!("sync_start_tonic_server end error msg: {:?}", e);
                SDKApiResult::error_string(e.to_string())
            } else {
                SDKApiResult::ok()
            }
        } else {
            if !set_tonic_server_start(false) {
                log::error!("sync_start_tonic_server exit runtime error set_tonic_server_start return false and is_connect_state_true = {:?}!",is_tonic_server_start());
            }
            SDKApiResult::from(anyhow!(SDKResultError::ErrorRuntime))
        }
    }

    async fn async_start_tonic_server(
        &self,
        server_host: String,
        subgroup_path: String,
        server_ak: String,
        server_secret: String,

        config_store_path: String,
        download_base_path: String,
        enable_http_callback: bool,

        receive_callback: Box<dyn P2PRPCServiceCallback>,
        turn_config: String,
    ) -> Result<(), anyhow::Error> {
        let mut rtm_config = P2PRPCConfig::from_json(&turn_config)?;

        rtm_config.device_package = if rtm_config.device_package == "" {
            "com.localhost.test".to_owned()
        } else {
            rtm_config.device_package
        };

        rtm_config.device_id = if rtm_config.device_id == "" {
            log::error!("device_id can not empty,default.device set!");
            format!("{}", "default.device")
        } else {
            rtm_config.device_id
        };

        rtm_config.user_id = if rtm_config.user_id == "" {
            format!("{}", "default.user")
        } else {
            rtm_config.user_id
        };

        let P2PRPCConfig {
            device_package,
            device_id,
            user_id,
            client_id,

            turn_uri,
            turn_usr,
            turn_pwd,
            turn_ttl,
        } = rtm_config;
        //request client
        let client = reqwest::Client::builder()
            .connect_timeout(Duration::from_secs(DEFAULT_CONNECTION_TIMEOUT_SECS))
            .gzip(true)
            // .trust_dns(true)
            .build()?;
        // config store path
        let config_file_saved_path = if config_store_path == "" {
            "./".to_owned()
        } else {
            let config_dir_gen_path = config_store_path.clone();
            let path_exit = match File::open(config_dir_gen_path.clone()) {
                Ok(_) => true,
                Err(err) => {
                    if std::io::ErrorKind::AlreadyExists == err.kind() {
                        true
                    } else {
                        false
                    }
                }
            };
            if !path_exit {
                fs::create_dir_all(config_dir_gen_path.clone()).unwrap_or_else(|why| {
                    log::error!(
                        "p2prpc config path {:?} mkdir failed {:?}",
                        config_dir_gen_path,
                        why.kind()
                    );
                });
            }
            config_dir_gen_path
        };

        let download_file_saved_path = if download_base_path == "" {
            "./".to_owned()
        } else {
            let download_dir_gen_path = download_base_path.clone();
            let path_exit = match File::open(download_dir_gen_path.clone()) {
                Ok(_) => true,
                Err(err) => {
                    if std::io::ErrorKind::AlreadyExists == err.kind() {
                        true
                    } else {
                        false
                    }
                }
            };
            if !path_exit {
                fs::create_dir_all(download_dir_gen_path.clone()).unwrap_or_else(|why| {
                    log::error!(
                        "p2prpc config path {:?} mkdir failed {:?}",
                        download_dir_gen_path,
                        why.kind()
                    );
                });
            }
            download_dir_gen_path
        };

        let config_store_path_buf: PathBuf = config_file_saved_path.clone().into();
        // config map
        let mut config = OptionalP2PRPCLocalStorageConfig::load(
            &APP_INFO,
            P2PRPC_LOCAL_CONFIG_FILE_NAME,
            &config_store_path_buf,
        )
        .map_err(|e| {
            log::warn!("load config err: {}", e.to_string());
            e
        })
        .unwrap_or(OptionalP2PRPCLocalStorageConfig::default());
        log::info!("p2prtc load config {:?}", config);

        let saved_device_host_key = format!(
            "{}-{}-{}-{}",
            server_host, device_id, device_package, user_id
        );

        let group_turn_host_map = config.group_turn_host_map.get_or_insert_default();
        let group_turn_user_map = config.group_turn_user_map.get_or_insert_default();
        let group_turn_pwd_map = config.group_turn_pwd_map.get_or_insert_default();

        let turn_addrs = self
            .refresh_turn_addr(
                &client,
                &server_host,
                &subgroup_path,
                &server_ak,
                &server_secret,
                &device_id,
                &device_package,
                &user_id,
            )
            .await?;

        group_turn_host_map.insert(
            saved_device_host_key.to_owned(),
            turn_addrs.turn_group_address.to_owned(),
        );
        group_turn_user_map.insert(
            saved_device_host_key.to_owned(),
            turn_addrs.turn_user_name.to_owned(),
        );
        group_turn_pwd_map.insert(
            saved_device_host_key.to_owned(),
            turn_addrs.turn_user_pwd.to_owned(),
        );

        // all tonic tunnel message call back,
        // include kcp tunnel,tonic server event.
        let callback: Box<dyn TonicTunnelCallback> =
            Box::new(CustomTonicServerCallback::new(receive_callback));

        let uri = URIInfo {
            username: turn_usr,
            uri: turn_uri,
            ttl: turn_ttl as u64,
            password: turn_pwd,
        };

        // step1:
        // create a tonic server.
        // create (service_callback_tx,service_callback_rx)<TunnelConnectMsg> future channel.
        // create (on_remote_content_tx,on_remote_content_rx)<send_content_msg::Msg> tokio channel.
        // create a loop to received service_callback_rx,on_remote_content_rx to callback to application.
        // then pass service_callback_tx to handler serve_with_incoming(incoming) messsage
        // pass on_remote_content_tx to get_or_set_remote_content_callback()

        // step2:
        // create (tunnel_tx, tunnel_rx)<TunnelConnectMsg> and pass tunnel_tx to TunnelCreatorImp::start_tonic_tunnel.
        // serve_with_incoming(incoming) wait for tunnel_rx then pass message to service_callback_tx.
        // TunnelCreatorImp::start_tonic_tunnel create tuple (adapter,signal_receiver:UnboundedSender<WsReceivedMessage>)signal-adapter to send and receive websocket messages,

        // and create signal_channel_map(tunnel_id,(tx<TunnelSignalIceMsg>,remote_id,state)) to handler received message from remote_client_id
        // and create std::thread::spawn thread map for tunnel_id
        // and create (create_tx, create_rx)<IceCommandMsg> future channel
        // and create (sender_tx,sender_rx)<(String, TunnelSignalIceMsg)> future channel
        // and create (tunnel_creator_callback_tx,tunnel_creator_callback_rx)<TunnelConnectMsg>
        // then tunnel_tx notify tunnel_rx TunnelCreatorImp ready,callback with create_tx, and sender_tx
        // pass tunnel_creator_callback_tx copy,signal_channel_map.get(tunnel_id) to step3:task2 every tunnel thread.

        // step3:
        // TunnelCreatorImp::start_tonic_tunnel create three future task:
        // task1:
        // signal_receiver_handler to loop signal_receiver handle signal message.
        // when instreated event received and is controlling,send IceCommandMsg by create_tx
        // task2:
        // command_handler to loop create_rx, when received IceCreateMsg,then spwan a thread:
        // and run async_new_tunnel_connect() to handle webrtc-ice connection.
        // task3:
        // send_callback_handler to loop tunnel_creator_callback_rx and sender_rx.
        // listen OnTunnelStateChange and if state changed to close then send to remote target and callback.
        let server = TonicServer::create_service_with_cb(
            client_id,
            callback,
            config_file_saved_path,
            uri,
            download_file_saved_path,
            enable_http_callback,
        )
        .await?;

        // wait here.
        server.join().await
    }

    fn stop_tonic_server(&self) -> Result<(), anyhow::Error> {
        let creator = get_tunnel_creator().ok_or(SDKResultError::ErrorNoTonicServerStarted)?;
        creator.stop_tonic_server()?;
        Ok(())
    }

    fn sync_connect_tonic_tunnel(
        &self,
        remote_client_id: String,
        tunnel_id: String,
        service_command_recv: UnboundedReceiver<TunnelServiceCommand>,
        client_command_recv: UnboundedReceiver<TunnelClientCommand>,
    ) -> SDKApiResult {
        if let Ok(rt) = Builder::new_multi_thread()
            .on_thread_start(|| {
                // call_attach_fn();
            })
            .on_thread_stop(|| {
                // call_detach_fn();
            })
            .enable_all()
            .build()
        {
            // block on tokio async loop. sync_connect must be called in a thread!
            let context = format!(
                "remote_target_id:{} tunnel_id:{} ",
                remote_client_id, tunnel_id
            );

            // block on received TunnelServiceCommand & TunnelClientCommand,
            // for tunnel_id
            let res = rt.block_on(self.async_tunnel_connect(
                remote_client_id,
                tunnel_id.to_owned(),
                service_command_recv,
                client_command_recv,
            ));

            let (serivce_sender_res, client_sender_res) = (
                get_or_set_service_command_sender(true, &tunnel_id, None),
                get_or_set_tunnel_client_command_sender(true, &tunnel_id, None),
            );
            log::info!("sync_connect_tonic_tunnel end {} res: {:?}", context, &res);
            if serivce_sender_res.is_err() || client_sender_res.is_err() {
                log::error!("sync_connect_tonic_tunnel _sender_res.is_err() || client_sender_res.is_err()! ");
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorSyncConnectTonicTunnel));
            }
            if let Err(err) = res {
                return SDKApiResult::from(err);
            } else {
                log::info!("sync_connect_tonic_tunnel exit normal! ");
                return SDKApiResult::ok();
            }
        } else {
            let context = format!(
                "remote_target_id:{} tunnel_id:{} ",
                remote_client_id, tunnel_id
            );
            let (serivce_sender_res, client_sender_res) = (
                get_or_set_service_command_sender(true, &tunnel_id, None),
                get_or_set_tunnel_client_command_sender(true, &tunnel_id, None),
            );
            if serivce_sender_res.is_err() || client_sender_res.is_err() {
                log::error!("sync_connect_tonic_tunnel _sender_res.is_err() || client_sender_res.is_err()! context = {}",context);
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorSyncConnectTonicTunnel));
            };
            SDKApiResult::from(anyhow!(SDKResultError::ErrorRuntime))
        }
    }

    async fn async_tunnel_connect(
        &self,
        // connect to remote tuple(client_id,tunnel_id).
        remote_client_id: String,
        tunnel_id: String,
        mut service_command_recv: UnboundedReceiver<TunnelServiceCommand>,
        mut client_command_recv: UnboundedReceiver<TunnelClientCommand>,
    ) -> Result<String, anyhow::Error> {
        let _tunnel_id = if tunnel_id == "" {
            id_generate_xbit(6)
        } else {
            tunnel_id
        };
        let tunnel_id = _tunnel_id.clone();

        // are there any content user callback tx?
        let user_tonic_srv_content_callback = get_or_set_remote_content_callback(false, None)
            .await
            .ok_or(anyhow::anyhow!(
                "user_tonic_srv_content_callback not set,please call sync_start_tonic_server first!",
            ))?;

        let user_tonic_srv_bytebuffer_callback = get_or_set_bytebuffer_callback(false, None)
            .await
            .ok_or(anyhow::anyhow!(
            "user_tonic_srv_bytebuffer_callback not set,please call sync_start_tonic_server first!",
        ))?;

        // listen from service command hander.
        let service_command_handler = async {
            log::info!("async_tunnel_connect get tonic client await!");
            // see async_connect_server step3:task2:
            // command_handler to loop create_rx, when received IceCreateMsg,then spwan a thread:
            // and run async_new_tunnel_connect() to handle webrtc-ice connection.
            // get_kcp_imp() return a tunnel_connection.
            let mut client = _get_tonic_client(&remote_client_id, &tunnel_id).await?;

            log::info!("async_tunnel_connect get tonic client");

            // let mut my_callback: Box<dyn TunnelFileTaskTrait> =
            //     Box::new(MyTunnelFileTaskCallback { callback });
            // wait from service_command_recv
            while let Some(command) = service_command_recv.next().await {
                log::info!("async_tunnel_connect command_recv.next()");
                match command {
                    // if received browse file ,with tuple(task_id,requst)
                    // TunnelServiceCommand::BrowseFile((task_id, request)) => {
                    //     log::info!("async_tunnel_connect ServiceCommand::BrowseFile");

                    //     // use client to request browser_file. this is rpc call.
                    //     let data = async {
                    //         let res = client.browser_file(request).await?.into_inner();
                    //         let data = serde_json::to_string(&res)?;
                    //         let res: Result<String, anyhow::Error> = Ok(data);
                    //         res
                    //     }
                    //     .await;

                    //     match data {
                    //         Ok(data) => {
                    //             my_callback.on_browse_result(task_id, true, data);
                    //         }
                    //         Err(e) => {
                    //             my_callback.on_browse_result(task_id, false, e.to_string());
                    //         }
                    //     }
                    // }
                    // // if received download file with download message
                    // TunnelServiceCommand::Download(msg) => {
                    //     log::info!("async_tunnel_connect ServiceCommand::Download");
                    //     // let my_callback = &mut my_callback as &mut Box<dyn TaskCallbackTrait>;
                    //     let DownloadMsg {
                    //         task_id,
                    //         save_file_path,
                    //         download_file_path,
                    //     } = msg;
                    //     // use client to request download file. this is rpc call.
                    //     if let Err(e) = download_file_with_trait(
                    //         task_id,
                    //         &mut client,
                    //         &download_file_path,
                    //         save_file_path,
                    //         &mut my_callback,
                    //     )
                    //     .await
                    //     {
                    //         log::error!("async_tunnel_connect err: {:?}", e);
                    //     }
                    // }
                    // if received send content start message
                    TunnelServiceCommand::SendContentStart((start, timestamp)) => {
                        log::info!("async_tunnel_connect TunnelServiceCommand::SendContentStart");
                        let start_tunnel_id = start.tunnel_id.clone();

                        if let Ok(sender_opt) =
                            get_or_set_content_sender(false, &start_tunnel_id, None)
                        {
                            if let Some(_sender) = sender_opt {
                                // already has sender ,and receiver a stale sent content message !
                                let current_timestamp = time_utils::timestamp_ms_i64();
                                if current_timestamp - timestamp > 5000 {
                                    log::error!("SendContentStart message is stale!,disconnect tunnel, {:?}",start_tunnel_id);
                                    // break while to accelector reconnect!
                                    if let Err(e) = user_tonic_srv_content_callback.send(Msg::Stop(
                                        SendContentStop {
                                            tunnel_id: tunnel_id.clone(),
                                            is_confirm: true,
                                        },
                                    )) {
                                        log::info!("SendContentStart message is stale handle_selected_contents_duplex outbound_user_callback_tx stop Err:{:?}", e);
                                    }
                                    break;
                                } else {
                                    log::warn!("SendContentStart message for tunnel {:?} already setted,ignore!",start_tunnel_id);
                                }
                            };
                        };

                        let res = handle_selected_contents_duplex(
                            start,
                            &mut client,
                            user_tonic_srv_content_callback.clone(),
                        )
                        .await;
                        // wait here for handle_selected_contents_duplex end.
                        match get_or_set_content_sender(true, &start_tunnel_id, None) {
                            Ok(sender) => {
                                if sender.is_some() {
                                    if let Err(e) = user_tonic_srv_content_callback.send(Msg::Stop(
                                        SendContentStop {
                                            tunnel_id: tunnel_id.clone(),
                                            is_confirm: true,
                                        },
                                    )) {
                                        log::info!("handle_selected_contents_duplex outbound_user_callback_tx stop Err:{:?}", e);
                                    }
                                }
                            }
                            Err(e) => {
                                log::info!(
                                    "handle_selected_contents_duplex get_or_set_content_sender Err:{:?}",
                                    e
                                );
                            }
                        }
                        log::info!(
                            "async_tunnel_connect ClientCommand::SendContentStart end res: {:?}",
                            res
                        );
                    }
                    TunnelServiceCommand::SendByteBufferStart((start, timestamp)) => {
                        log::info!(
                            "async_tunnel_connect TunnelServiceCommand::SendByteBufferStart"
                        );
                        let start_tunnel_id = start.tunnel_id.clone();

                        if let Ok(sender_opt) =
                            get_or_set_bytebuffer_sender(false, &start_tunnel_id, None)
                        {
                            if let Some(_sender) = sender_opt {
                                // already has sender ,and receiver a stale sent content message !
                                let current_timestamp = time_utils::timestamp_ms_i64();
                                if current_timestamp - timestamp > 5000 {
                                    log::error!("SendByteBufferStart message is stale!,disconnect tunnel, {:?}",start_tunnel_id);
                                    // break while to accelector reconnect!
                                    if let Err(e) = user_tonic_srv_bytebuffer_callback.send(
                                        Bfmsg::BfStop(ByteBufferStop {
                                            tunnel_id: tunnel_id.clone(),
                                            is_confirm: true,
                                        }),
                                    ) {
                                        log::info!("SendByteBufferStart message is stale handle_selected_bytebuffer_duplex outbound_user_callback_tx stop Err:{:?}", e);
                                    }
                                    break;
                                } else {
                                    log::warn!("SendByteBufferStart message for tunnel {:?} already setted,ignore!",start_tunnel_id);
                                }
                            };
                        };

                        let res = handle_selected_bytebuffer_duplex(
                            start,
                            &mut client,
                            user_tonic_srv_bytebuffer_callback.clone(),
                        )
                        .await;
                        // wait here for handle_selected_bytebuffer_duplex end.
                        match get_or_set_bytebuffer_sender(true, &start_tunnel_id, None) {
                            Ok(sender) => {
                                if sender.is_some() {
                                    if let Err(e) = user_tonic_srv_bytebuffer_callback.send(
                                        Bfmsg::BfStop(ByteBufferStop {
                                            tunnel_id: tunnel_id.clone(),
                                            is_confirm: true,
                                        }),
                                    ) {
                                        log::info!("handle_selected_bytebuffer_duplex outbound_user_callback_tx stop Err:{:?}", e);
                                    }
                                }
                            }
                            Err(e) => {
                                log::info!(
                                    "handle_selected_bytebuffer_duplex get_or_set_bytebuffer_sender Err:{:?}",
                                    e
                                );
                            }
                        }
                        log::info!(
                            "async_tunnel_connect ClientCommand::SendByteBufferStart end res: {:?}",
                            res
                        );
                    }
                }
                log::info!("async_tunnel_connect command_recv.next() finished");
            }
            let res: Result<(), anyhow::Error> = Ok(());
            res
        };

        let client_command_handler = async {
            while let Some(command) = client_command_recv.next().await {
                log::info!("async_tunnel_connect client_command_recv.next()");
                match command {
                    TunnelClientCommand::StopConnect => {
                        log::info!("async_tunnel_connect ClientCommand::StopConnect");

                        match get_or_set_content_sender(true, &tunnel_id, None) {
                            Ok(sender) => {
                                if sender.is_some() {
                                    if let Err(e) = user_tonic_srv_content_callback.send(Msg::Stop(
                                        SendContentStop {
                                            tunnel_id: tunnel_id.clone(),
                                            is_confirm: true,
                                        },
                                    )) {
                                        log::info!("handle_send_content user_tonic_server_callback stop Err:{:?}", e);
                                    }
                                }
                            }
                            Err(e) => {
                                log::info!(
                                    "handle_send_content get_or_set_content_sender Err:{:?}",
                                    e
                                );
                            }
                        }
                        break;
                    }
                    TunnelClientCommand::StopByteBuffer => {
                        log::info!("async_tunnel_connect ClientCommand::StopByteBuffer");

                        match get_or_set_bytebuffer_sender(true, &tunnel_id, None) {
                            Ok(sender) => {
                                if sender.is_some() {
                                    if let Err(e) = user_tonic_srv_bytebuffer_callback.send(
                                        Bfmsg::BfStop(ByteBufferStop {
                                            tunnel_id: tunnel_id.clone(),
                                            is_confirm: true,
                                        }),
                                    ) {
                                        log::info!("handle_bytebuffer_content user_tonic_srv_bytebuffer_callback stop Err:{:?}", e);
                                    }
                                }
                            }
                            Err(e) => {
                                log::info!(
                                    "handle_bytebuffer_content get_or_set_bytebuffer_sender Err:{:?}",
                                    e
                                );
                            }
                        }
                        break;
                    }
                }
            }
            let res: Result<(), anyhow::Error> = Ok(());
            res
        };
        pin_mut!(service_command_handler, client_command_handler);

        // wait for future::select branch finish!
        let res =
            match futures_util::future::select(service_command_handler, client_command_handler)
                .await
            {
                Either::Left((value1, _)) => value1,
                Either::Right((value2, _)) => value2,
            };

        if let Some(mut creator) = get_tunnel_creator() {
            if let Err(e) = creator.disconnect(&_tunnel_id) {
                log::error!("creator.disconnect err: {:?}", e);
            }
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await; //wait for close

        log::info!(
            "pub async fn async_tunnel_connect end tunnel_id: {}",
            _tunnel_id
        );

        res?;

        Ok(_tunnel_id)
    }

    // close tonic tunnel.
    fn drop_tonic_tunnel(&self, tunnel_id: String) -> SDKApiResult {
        let result = close_remote_tonic_tunnel(tunnel_id);
        match result {
            Ok(_) => {
                return SDKApiResult::ok();
            }
            Err(err) => {
                log::error!(
                    "drop_tonic_tunnel call close_remote_tonic_tunnel error:{}",
                    err
                );
                return SDKApiResult::from(err);
            }
        }
    }

    // select tonic tunnel as content stream
    fn select_tonic_tunnel_as_content_stream(&self, tunnel_id: String) -> SDKApiResult {
        let result = select_content_service_duplex(tunnel_id);
        match result {
            Ok(_) => {
                return SDKApiResult::ok();
            }
            Err(err) => {
                log::error!(
                    "select_tonic_tunnel_as_content_stream call start_send_content_service error:{}",
                    err
                );
                return SDKApiResult::from(err);
            }
        }
    }

    // recall tonic tunnel as content stream
    fn recall_tonic_tunnel_as_content_stream(&self, tunnel_id: String) -> SDKApiResult {
        let result = cancel_content_service_duplex(tunnel_id, false);
        match result {
            Ok(_) => {
                return SDKApiResult::ok();
            }
            Err(err) => {
                log::error!(
                    "cancel_content_service_duplex call stop_send_content_service error:{}",
                    err
                );
                return SDKApiResult::from(err);
            }
        }
    }

    // select tonic tunnel as bytebuf stream
    fn select_tonic_tunnel_as_bytebuf_stream(&self, tunnel_id: String) -> SDKApiResult {
        let result = select_bytebuf_service_duplex(tunnel_id);
        match result {
            Ok(_) => {
                return SDKApiResult::ok();
            }
            Err(err) => {
                log::error!(
                    "select_tonic_tunnel_as_bytebuf_stream call start_send_bytebuf_service error:{}",
                    err
                );
                return SDKApiResult::from(err);
            }
        }
    }

    // recall tonic tunnel as bytebuf stream
    fn recall_tonic_tunnel_as_bytebuf_stream(&self, tunnel_id: String) -> SDKApiResult {
        let result = cancel_bytebuf_service_duplex(tunnel_id, false);
        match result {
            Ok(_) => {
                return SDKApiResult::ok();
            }
            Err(err) => {
                log::error!(
                    "recall_tonic_tunnel_as_bytebuf_stream call stop_send_bytebuf_service error:{}",
                    err
                );
                return SDKApiResult::from(err);
            }
        }
    }

    fn send_text_content(
        &self,
        tunnel_id: String,
        id: String,
        event: String,
        data: String,
        can_drop: bool,
    ) -> SDKApiResult {
        let result = request_send_content(tunnel_id, id, event, data, can_drop);
        match result {
            Ok(id) => {
                return SDKApiResult::msg(id);
            }
            Err(err) => {
                log::error!("send_text_content call request_send_content error:{}", err);
                return SDKApiResult::from(err);
            }
        }
    }

    fn send_bytebuffer_content(
        &self,
        tunnel_id: String,
        id: String,
        can_drop: bool,
        video_buf: Vec<u8>,
        audio_buf: Vec<u8>,
    ) -> SDKApiResult {
        let result = request_send_bytes(tunnel_id, id, can_drop, video_buf, audio_buf);
        match result {
            Ok(id) => {
                return SDKApiResult::msg(id);
            }
            Err(err) => {
                log::error!("request_send_bytebuffer call error:{}", err);
                return SDKApiResult::from(err);
            }
        }
    }

    fn refresh_candidate(&self, is_online: bool) -> SDKApiResult {
        if is_online {
            let result = request_gather_candidates();
            match result {
                Ok(_) => {
                    return SDKApiResult::ok();
                }
                Err(err) => {
                    log::error!(
                        "on_network_change call request_gather_candidates error:{}",
                        err
                    );
                    return SDKApiResult::from(err);
                }
            }
        }

        SDKApiResult::ok()
    }

    fn soft_link_file_mapping(&self, operate: u8, file_path: String) -> SDKApiResult {
        let result = operate_download_path_map(operate, file_path);
        match result {
            Ok(id) => {
                return SDKApiResult::msg(id);
            }
            Err(err) => {
                log::error!(
                    "soft_link_file_mapping call operate_download_path_map error:{}",
                    err
                );
                return SDKApiResult::from(err);
            }
        }
    }

    fn is_remote_not_need_reply(&self, remote_id: String) -> SDKApiResult {
        if pairinlanmap::is_remote_in_not_need_reply(remote_id.to_owned()) {
            SDKApiResult::ok()
        } else {
            SDKApiResult::error_string("false".to_owned())
        }
    }
}

impl P2PRPCServiceContract for P2PRPCServiceContractImp {
    fn init_p2prpc_sdk(tag: String, log_level: String) -> SDKApiResult {
        if set_sdk_instance(P2PRPCServiceContractImp::new()) {
            logutil::global_log_init(tag.as_str(), log_level.as_str());
        }
        SDKApiResult::ok()
    }

    /// loop_block_on_start_p2prpc_server: loop start p2prtc server, loop block means infinity run or unwrap inner then break.
    fn loop_block_on_start_p2prpc_server(
        sg_host: String,          // gateway query clientid url host address
        sg_subgroup_path: String, //  gateway
        sg_access_key: String,    // gateway registed http request auth access key
        sg_access_secret: String, // gateway registed http request auth access secret,corresponding for access key

        config_file_saved_path: String,    // config file saved path
        download_files_saved_path: String, // download files saved path
        enable_httpd_callback: bool,       // check if enable httpd callback
        p2prpc_callback: Box<dyn P2PRPCServiceCallback>, //p2prpc call back dyn point
        ice_turn_config_json: String,      // ice turn config json string, mapping to TurnAddrs
    ) -> SDKApiResult {
        // safe set state in multithread.
        if !set_tonic_server_start(true) {
            log::warn!(
                "loop_block_on_start_p2prpc_server called,but tonic server already started,please handle the 40012 error and ignore this error result!"
            );
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorTonicServerAlreadyStarted));
        }

        match P2PRTC_INSTANCE.try_get() {
            Some(instance) => {
                return instance.read().unwrap().sync_start_tonic_server(
                    sg_host,
                    sg_subgroup_path,
                    sg_access_key,
                    sg_access_secret,
                    config_file_saved_path,
                    download_files_saved_path,
                    enable_httpd_callback,
                    p2prpc_callback,
                    ice_turn_config_json,
                );
            }
            None => {
                if !set_tonic_server_start(false) {
                    log::error!("sync_start_tonic_server exit RTM_SDK_INSTANCE.get().read() error set_tonic_server_start return false and is_tonic_server_start = {:?}!",is_tonic_server_start());
                }
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorRtmSdkNotInited));
            }
        };
    }

    /// stop p2prpc server
    fn stop_p2prpc_server() -> SDKApiResult {
        match P2PRTC_INSTANCE.try_get() {
            Some(instance) => {
                if !is_tonic_server_start() {
                    return SDKApiResult::from(anyhow!(SDKResultError::ErrorNoTonicServerStarted));
                }

                if let Err(err) = instance.read().unwrap().stop_tonic_server() {
                    return SDKApiResult::from(err);
                };
                return SDKApiResult::ok();
            }
            None => {
                if !set_tonic_server_start(false) {
                    log::error!("stop_p2prpc_server exit P2PRTC_INSTANCE.get().read() error set_tonic_server_start return false and is_tonic_server_start = {:?}!",is_tonic_server_start());
                }
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorRtmSdkNotInited));
            }
        };
    }

    /// block_on_make_p2prpc_tunnel:  loop make p2prtc tunnel, loop block means infinity run or unwrap inner then break.
    fn block_on_make_p2prpc_tunnel(
        local_client_id: String,  //local_client_id of this p2prpc pair tunnel
        remote_client_id: String, // remote_device_id of this p2prpc pair tunnel
        tunnel_id: String,        // p2prpc tunnel of unique tunnel id
    ) -> SDKApiResult {
        // check sender is set ,support multithread.
        let (serivce_sender_res, client_sender_res) = (
            get_or_set_service_command_sender(false, &tunnel_id, None),
            get_or_set_tunnel_client_command_sender(false, &tunnel_id, None),
        );
        if serivce_sender_res.is_err() || client_sender_res.is_err() {
            log::error!("loop_block_on_make_p2prpc_tunnel set none. _sender_res.is_err() || client_sender_res.is_err() ");
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorSyncConnectTonicTunnel));
        }
        if serivce_sender_res.unwrap().is_some() || client_sender_res.unwrap().is_some() {
            log::error!("loop_block_on_make_p2prpc_tunnel _sender_res already connect error!");
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorAlreadyConnectTonicTunnelID));
        }

        let (service_command_tx, service_command_rx) = futures_channel::mpsc::unbounded();
        let (client_command_tx, client_command_rx) = futures_channel::mpsc::unbounded();

        let serivce_sender_res = get_or_set_service_command_sender(
            true,
            &tunnel_id,
            Some((
                service_command_tx,
                local_client_id.to_owned(),
                remote_client_id.to_owned(),
            )),
        );

        let client_sender_res =
            get_or_set_tunnel_client_command_sender(true, &tunnel_id, Some(client_command_tx));

        if serivce_sender_res.is_err() || client_sender_res.is_err() {
            log::error!("loop_block_on_make_p2prpc_tunnel set some. _sender_res.is_err() || client_sender_res.is_err() ");
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorSyncConnectTonicTunnel));
        }

        match P2PRTC_INSTANCE.try_get() {
            Some(instance) => {
                return instance.read().unwrap().sync_connect_tonic_tunnel(
                    remote_client_id,
                    tunnel_id,
                    service_command_rx,
                    client_command_rx,
                );
            }
            None => {
                let (serivce_sender_res, client_sender_res) = (
                    get_or_set_service_command_sender(true, &tunnel_id, None),
                    get_or_set_tunnel_client_command_sender(true, &tunnel_id, None),
                );
                if serivce_sender_res.is_err() || client_sender_res.is_err() {
                    log::error!("loop_block_on_make_p2prpc_tunnel _sender_res.is_err() || client_sender_res.is_err() ");
                    return SDKApiResult::from(anyhow!(
                        SDKResultError::ErrorSyncConnectTonicTunnel
                    ));
                }
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorRtmSdkNotInited));
            }
        };
    }

    /// drop the p2prpc tunnel
    fn drop_p2prpc_tunnel(tunnel_id: String, // p2prpc tunnel of unique tunnel id
    ) -> SDKApiResult {
        let (serivce_sender_res, client_sender_res) = (
            get_or_set_service_command_sender(false, &tunnel_id, None),
            get_or_set_tunnel_client_command_sender(false, &tunnel_id, None),
        );
        if serivce_sender_res.is_err() || client_sender_res.is_err() {
            log::error!(
                "close_tonic_tunnel set none. _sender_res.is_err() || client_sender_res.is_err() "
            );
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorTonicTunnelChannelBroken));
        }
        if serivce_sender_res.unwrap().is_none() && client_sender_res.unwrap().is_none() {
            log::error!(
                "close_tonic_tunnel set none. _sender_res.is_err() || client_sender_res.is_err() "
            );
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorNotFoundTonicTunnelID));
        }
        match P2PRTC_INSTANCE.try_get() {
            Some(instance) => {
                return instance.read().unwrap().drop_tonic_tunnel(tunnel_id);
            }
            None => {
                let (serivce_sender_res, client_sender_res) = (
                    get_or_set_service_command_sender(true, &tunnel_id, None),
                    get_or_set_tunnel_client_command_sender(true, &tunnel_id, None),
                );
                if serivce_sender_res.is_err() || client_sender_res.is_err() {
                    log::error!(
                        "close_tonic_tunnel _sender_res.is_err() || client_sender_res.is_err() "
                    );
                    return SDKApiResult::from(anyhow!(
                        SDKResultError::ErrorSyncConnectTonicTunnel
                    ));
                }
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorRtmSdkNotInited));
            }
        };
    }

    /// select the p2prpc tunnel as duplex text tunnel
    fn select_text_tunnel_duplex(
        tunnel_id: String, // p2prpc tunnel of unique tunnel id
    ) -> SDKApiResult {
        let (serivce_sender_res, client_sender_res) = (
            get_or_set_service_command_sender(false, &tunnel_id, None),
            get_or_set_tunnel_client_command_sender(false, &tunnel_id, None),
        );
        if serivce_sender_res.is_err() || client_sender_res.is_err() {
            log::error!("select_text_tunnel_duplex set none. _sender_res.is_err() || client_sender_res.is_err() ");
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorTonicTunnelChannelBroken));
        }
        if serivce_sender_res.unwrap().is_none() && client_sender_res.unwrap().is_none() {
            log::error!("select_text_tunnel_duplex set none. _sender_res.is_err() || client_sender_res.is_err() ");
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorNotFoundTonicTunnelID));
        }
        match P2PRTC_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .select_tonic_tunnel_as_content_stream(tunnel_id);
            }
            None => {
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorRtmSdkNotInited));
            }
        };
    }

    /// cancel the p2prpc tunnel as duplex text tunnel
    fn cancel_text_tunnel_duplex(
        tunnel_id: String, // p2prpc tunnel of unique tunnel id
    ) -> SDKApiResult {
        let (serivce_sender_res, client_sender_res) = (
            get_or_set_service_command_sender(false, &tunnel_id, None),
            get_or_set_tunnel_client_command_sender(false, &tunnel_id, None),
        );
        if serivce_sender_res.is_err() || client_sender_res.is_err() {
            log::error!("cancel_text_tunnel_duplex set none. _sender_res.is_err() || client_sender_res.is_err() ");
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorTonicTunnelChannelBroken));
        }
        if serivce_sender_res.unwrap().is_none() && client_sender_res.unwrap().is_none() {
            log::error!("cancel_text_tunnel_duplex set none. _sender_res.is_err() || client_sender_res.is_err() ");
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorNotFoundTonicTunnelID));
        }
        match P2PRTC_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .recall_tonic_tunnel_as_content_stream(tunnel_id);
            }
            None => {
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorRtmSdkNotInited));
            }
        };
    }

    /// select the p2prpc tunnel as duplex bytebuffer tunnel
    fn select_bytebuf_tunnel_duplex(
        tunnel_id: String, // p2prpc tunnel of unique tunnel id
    ) -> SDKApiResult {
        let (serivce_sender_res, client_sender_res) = (
            get_or_set_service_command_sender(false, &tunnel_id, None),
            get_or_set_tunnel_client_command_sender(false, &tunnel_id, None),
        );
        if serivce_sender_res.is_err() || client_sender_res.is_err() {
            log::error!("select_bytebuf_tunnel_duplex set none. _sender_res.is_err() || client_sender_res.is_err() ");
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorTonicTunnelChannelBroken));
        }
        if serivce_sender_res.unwrap().is_none() && client_sender_res.unwrap().is_none() {
            log::error!("select_bytebuf_tunnel_duplex set none. _sender_res.is_err() || client_sender_res.is_err() ");
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorNotFoundTonicTunnelID));
        }
        match P2PRTC_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .select_tonic_tunnel_as_bytebuf_stream(tunnel_id);
            }
            None => {
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorRtmSdkNotInited));
            }
        };
    }

    /// cancel the p2prpc tunnel as duplex bytebuffer tunnel
    fn cancel_bytebuf_tunnel_duplex(
        tunnel_id: String, // p2prpc tunnel of unique tunnel id
    ) -> SDKApiResult {
        let (serivce_sender_res, client_sender_res) = (
            get_or_set_service_command_sender(false, &tunnel_id, None),
            get_or_set_tunnel_client_command_sender(false, &tunnel_id, None),
        );
        if serivce_sender_res.is_err() || client_sender_res.is_err() {
            log::error!("cancel_bytebuf_tunnel_duplex set none. _sender_res.is_err() || client_sender_res.is_err() ");
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorTonicTunnelChannelBroken));
        }
        if serivce_sender_res.unwrap().is_none() && client_sender_res.unwrap().is_none() {
            log::error!("cancel_bytebuf_tunnel_duplex set none. _sender_res.is_err() || client_sender_res.is_err() ");
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorNotFoundTonicTunnelID));
        }
        match P2PRTC_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .recall_tonic_tunnel_as_bytebuf_stream(tunnel_id);
            }
            None => {
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorRtmSdkNotInited));
            }
        };
    }

    /// send message with p2prpc text tunnel
    fn send_text_message(
        tunnel_id: String, // p2prpc tunnel of unique tunnel id
        id: String,        // the unique message id
        event: String,     // message event name
        data: String,      // message data
        can_drop: bool,    // can drop means this message can drop when network traffic busy
    ) -> SDKApiResult {
        match P2PRTC_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .send_text_content(tunnel_id, id, event, data, can_drop);
            }
            None => {
                let (serivce_sender_res, client_sender_res) = (
                    get_or_set_service_command_sender(true, &tunnel_id, None),
                    get_or_set_tunnel_client_command_sender(true, &tunnel_id, None),
                );
                if serivce_sender_res.is_err() || client_sender_res.is_err() {
                    log::error!(
                        "send_text_message _sender_res.is_err() || client_sender_res.is_err() "
                    );
                    return SDKApiResult::from(anyhow!(
                        SDKResultError::ErrorSyncConnectTonicTunnel
                    ));
                }
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorRtmSdkNotInited));
            }
        };
    }

    /// send bytebuffer with p2prpc bytebuffer channel
    fn send_bytebuffer(
        tunnel_id: String,  // p2prpc tunnel of unique tunnel id
        id: String,         // the unique message id
        video_buf: Vec<u8>, // the video buffer vec
        audio_buf: Vec<u8>, // the audio buffer vec
        can_drop: bool,     //  can drop means this bytebuffer can drop when network traffic busy
    ) -> SDKApiResult {
        match P2PRTC_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .send_bytebuffer_content(tunnel_id, id, can_drop, video_buf, audio_buf);
            }
            None => {
                let (serivce_sender_res, client_sender_res) = (
                    get_or_set_service_command_sender(true, &tunnel_id, None),
                    get_or_set_tunnel_client_command_sender(true, &tunnel_id, None),
                );
                if serivce_sender_res.is_err() || client_sender_res.is_err() {
                    log::error!(
                        "send_bytebuffer _sender_res.is_err() || client_sender_res.is_err() "
                    );
                    return SDKApiResult::from(anyhow!(
                        SDKResultError::ErrorSyncConnectTonicTunnel
                    ));
                }
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorRtmSdkNotInited));
            }
        };
    }

    /// network_change_refresh_candidate:notify rpc tunnel device network drop or online
    fn network_change_refresh_candidate(
        is_online: bool, //notify network is online or offline
    ) -> SDKApiResult {
        match P2PRTC_INSTANCE.try_get() {
            Some(instance) => {
                return instance.read().unwrap().refresh_candidate(is_online);
            }
            None => {
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorRtmSdkNotInited));
            }
        };
    }

    /// notify_netwosoft_link_file_mapping:
    fn soft_link_file_mapping(
        add_or_remove: u8, // add or remove local file soft link to rpc mapping. 1 = add,0 = remove
        file_path: String, // local file abstract path.
    ) -> SDKApiResult {
        match P2PRTC_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .soft_link_file_mapping(add_or_remove, file_path);
            }
            None => {
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorRtmSdkNotInited));
            }
        };
    }

    /// is_p2prpc_remote_in_lan: check if the p2pprc tunnel pair is on the same lan.
    fn is_p2prpc_remote_in_lan(
        tunnel_id: String, // p2prpc tunnel of unique tunnel id
        remote_id: String, // p2prpc pair of remote id.
    ) -> SDKApiResult {
        let (serivce_sender_res, client_sender_res) = (
            get_or_set_service_command_sender(false, &tunnel_id, None),
            get_or_set_tunnel_client_command_sender(false, &tunnel_id, None),
        );
        if serivce_sender_res.is_err() || client_sender_res.is_err() {
            log::error!(
                "is_p2prpc_remote_in_lan serivce_sender_res set none. _sender_res.is_err() || client_sender_res.is_err() "
            );
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorTonicTunnelChannelBroken));
        }
        if serivce_sender_res.unwrap().is_none() && client_sender_res.unwrap().is_none() {
            log::error!(
                "is_p2prpc_remote_in_lan serivce_sender_res set none. _sender_res.is_err() || client_sender_res.is_err() "
            );
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorNotFoundTonicTunnelID));
        }

        match P2PRTC_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .is_remote_not_need_reply(remote_id.to_owned());
            }
            None => {
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorRtmSdkNotInited));
            }
        };
    }
}
